package com.ruirenyun.mq.receive;

import com.rabbitmq.client.Channel;
import com.ruirenyun.mq.config.RabbitConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;

@Component
@Slf4j
public class MessageReceive implements ChannelAwareMessageListener {

    /*@RabbitListener(bindings ={@QueueBinding(value = @Queue(value = RabbitConfig.FANOUT_QUEUE_NAME, durable = "true"),exchange = @Exchange(value = RabbitConfig.TEST_FANOUT_EXCHANGE, type = "fanout"))})
    @RabbitHandler
    public void processFanoutMsg(Message massage) {
        String msg = new String(massage.getBody(), StandardCharsets.UTF_8);
        log.info("############################ received Fanout message : " + msg);
    }

    @RabbitListener(bindings ={@QueueBinding(value = @Queue(value = RabbitConfig.FANOUT_QUEUE_NAME1, durable = "true"),exchange = @Exchange(value = RabbitConfig.TEST_FANOUT_EXCHANGE, type = "fanout"))})
    @RabbitHandler
    public void processFanout1Msg(Message massage) {
        String msg = new String(massage.getBody(), StandardCharsets.UTF_8);
        log.info("############################ received Fanout1 message : " + msg);
    }

    @RabbitListener(bindings ={@QueueBinding(value = @Queue(value = RabbitConfig.DIRECT_QUEUE_NAME, durable = "true"),exchange = @Exchange(value = RabbitConfig.TEST_DIRECT_EXCHANGE), key = RabbitConfig.DIRECT_ROUTINGKEY)})
    @RabbitHandler
    public void processDirectMsg(Message massage) {
        String msg = new String(massage.getBody(), StandardCharsets.UTF_8);
        //BenefitAddReqVo vo = JSON.parseObject(msg, BenefitAddReqVo.class);
        //System.out.println(vo.toString());
        log.info("############################ received Direct message : " + msg);
    }

    @RabbitListener(bindings ={@QueueBinding(value = @Queue(value = RabbitConfig.TOPIC_QUEUE_NAME, durable = "true"), exchange = @Exchange(value = RabbitConfig.TOPIC_EXCHANGE, type = "topic"), key = RabbitConfig.TOPIC_ROUTINGKEY)})
    @RabbitHandler
    public void processTopicMsg(Message massage) {
        String msg = new String(massage.getBody(), StandardCharsets.UTF_8);
        log.info("############################ received Topic message : " + msg);
    }*/

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            String consumerQueue = message.getMessageProperties().getConsumerQueue();
            String routingKey = message.getMessageProperties().getReceivedRoutingKey();
            if ((RabbitConfig.DIRECT_QUEUE_NAME).equals(consumerQueue)){
                String msg = new String(message.getBody(), StandardCharsets.UTF_8);
                log.info("############################ 消费的消息来自的队列名为：" + message.getMessageProperties().getConsumerQueue());
                log.info("############################ 消息成功消费到  direct_message:" + msg);
                log.info("############################ 执行test_direct_queue中的消息的业务处理流程......");
            }
            if ((RabbitConfig.FANOUT_QUEUE_NAME).equals(consumerQueue)){
                String msg = new String(message.getBody(), StandardCharsets.UTF_8);
                log.info("############################ 消费的消息来自的队列名为：" + message.getMessageProperties().getConsumerQueue());
                log.info("############################ 消息成功消费到  fanout_message:" + msg);
                log.info("############################ 执行test_fanout_queue中的消息的业务处理流程......");
            }
            if ((RabbitConfig.TOPIC_QUEUE_NAME).equals(consumerQueue) && (RabbitConfig.TOPIC_ROUTINGKEY).equals(routingKey)){
                String msg = new String(message.getBody(), StandardCharsets.UTF_8);
                log.info("############################ 消费的消息来自的队列名为：" + message.getMessageProperties().getConsumerQueue());
                log.info("############################ 消息成功消费到  topic_message:" + msg);
                log.info("############################ 执行test_topic_queue中的消息的业务处理流程......");
            }
            channel.basicAck(deliveryTag, true);
            //为true会重新放回队列
            //channel.basicReject(deliveryTag, true);
        }catch(Exception e){
            channel.basicReject(deliveryTag,true);
            e.printStackTrace();
        }
    }
}
